package org.go.threadpool;

import java.util.AbstractCollection;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.RunnableScheduledFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.go.trigger.Calendar;
import org.go.trigger.CalendarContent;
import org.go.trigger.CronExpression;
import org.go.trigger.Trigger;

/**
 * 
 * @author hejie
 * @see ScheduledThreadPoolExecutor
 */
public class MyScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService {
	static class MyDelayedWorkQueue extends AbstractCollection<Runnable> implements BlockingQueue<Runnable> {

		final DelayQueue<RunnableScheduledFuture> dq = new DelayQueue<RunnableScheduledFuture>();

		@Override
		public boolean add(Runnable x) {
			return dq.add((RunnableScheduledFuture) x);
		}

		@Override
		public void clear() {
			dq.clear();
		}

		@Override
		public boolean contains(Object x) {
			return dq.contains(x);
		}

		@Override
		public int drainTo(Collection<? super Runnable> c) {
			return dq.drainTo(c);
		}

		@Override
		public int drainTo(Collection<? super Runnable> c, int maxElements) {
			return dq.drainTo(c, maxElements);
		}

		@Override
		public Runnable element() {
			return dq.element();
		}

		@Override
		public boolean isEmpty() {
			return dq.isEmpty();
		}

		@Override
		public Iterator<Runnable> iterator() {
			return new Iterator<Runnable>() {
				private Iterator<RunnableScheduledFuture> it = dq.iterator();

				@Override
				public boolean hasNext() {
					return it.hasNext();
				}

				@Override
				public Runnable next() {
					return it.next();
				}

				@Override
				public void remove() {
					it.remove();
				}
			};
		}

		@Override
		public boolean offer(Runnable x) {
			return dq.offer((RunnableScheduledFuture) x);
		}

		@Override
		public boolean offer(Runnable x, long timeout, TimeUnit unit) {
			return dq.offer((RunnableScheduledFuture) x, timeout, unit);
		}

		@Override
		public Runnable peek() {
			return dq.peek();
		}

		@Override
		public Runnable poll() {
			return dq.poll();
		}

		@Override
		public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
			return dq.poll(timeout, unit);
		}

		@Override
		public void put(Runnable x) {
			dq.put((RunnableScheduledFuture) x);
		}

		@Override
		public int remainingCapacity() {
			return dq.remainingCapacity();
		}

		@Override
		public Runnable remove() {
			return dq.remove();
		}

		@Override
		public boolean remove(Object x) {
			return dq.remove(x);
		}

		@Override
		public int size() {
			return dq.size();
		}

		@Override
		public Runnable take() throws InterruptedException {
			//System.out.println("dq.size(): " + dq.size());
			//System.out.println("take");
			Runnable run = dq.take();
			//System.out.println("take: " + run.getClass().getName());
			return run;
		}

		@Override
		public Object[] toArray() {
			return dq.toArray();
		}

		@Override
		public <T> T[] toArray(T[] array) {
			return dq.toArray(array);
		}
	}

	public class GoTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
		/**
		 * Period in nanoseconds for repeating tasks.  A positive
		 * value indicates fixed-rate execution.  A negative value
		 * indicates fixed-delay execution.  A value of 0 indicates a
		 * non-repeating task.
		 */
		private final long period;

		/** Sequence number to break ties FIFO */
		private AtomicLong repeatCount = null;

		/** Sequence number to break ties FIFO */
		private final long sequenceNumber;

		/** The time the task is enabled to execute in nanoTime units */
		private long time;

		/** The actual task to be re-enqueued by reExecutePeriodic */
		RunnableScheduledFuture<V> outerTask = this;

		/**
		 * Creates a one-shot action with given nanoTime-based trigger.
		 */
		public GoTask(Callable<V> callable, long ns) {
			super(callable);
			this.time = ns;
			this.period = 0;
			this.sequenceNumber = sequencer.getAndIncrement();
		}

		/**
		* Creates a one-shot action with given nanoTime-based trigger time.
		*/
		public GoTask(Runnable r, V result, long ns) {
			super(r, result);
			this.time = ns;
			this.period = 0;
			this.sequenceNumber = sequencer.getAndIncrement();
		}

		/**
		 * Creates a periodic action with given nano time and period.
		 */
		public GoTask(Runnable r, V result, long ns, long period) {
			super(r, result);
			this.time = ns;
			this.period = period;
			this.sequenceNumber = sequencer.getAndIncrement();
		}

		/**
		 * Runs a periodic task.
		 */
		@SuppressWarnings("synthetic-access")
		private void runPeriodic() {
			boolean ok = super.runAndReset();
			boolean down = isShutdown();
			// Reschedule if not cancelled and not shutdown or policy allows
			if (ok && (!down || (getContinueExistingPeriodicTasksAfterShutdownPolicy() && !isTerminating()))) {
				long p = period;
				if (p > 0) {
					time += p;
				} else {
					time = now() - p;
				}
				MyScheduledThreadPoolExecutor.super.getQueue().add(this);
			}
			// This might have been the final executed delayed
			// task.  Wake up threads to check.
			else if (down) {

			}
			//interruptIdleWorkers();
		}

		@Override
		public int compareTo(Delayed other) {
			//System.out.println("a");
			if (other == this) // compare zero ONLY if same object
				return 0;
			if (other instanceof GoTask) {
				GoTask<?> x = (GoTask<?>) other;
				long diff = time - x.time;
				if (diff < 0)
					return -1;
				else if (diff > 0)
					return 1;
				else if (sequenceNumber < x.sequenceNumber)
					return -1;
				else
					return 1;
			}
			long d = (getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS));
			return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
		}

		@Override
		public long getDelay(TimeUnit unit) {
			long delay = unit.convert(time - now(), TimeUnit.NANOSECONDS);
			//System.out.println("delay:" + delay);
			return delay;
		}

		@Override
		public boolean isPeriodic() {
			return period != 0;
		}

		@SuppressWarnings("synthetic-access")
		@Override
		public void run() {
			//System.out.println("b");
			if (isPeriodic())
				runPeriodic();
			else
				GoTask.super.run();
		}
	}

	public class GoTask1<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
		/**
		 * Period in nanoseconds for repeating tasks.  A positive
		 * value indicates fixed-rate execution.  A negative value
		 * indicates fixed-delay execution.  A value of 0 indicates a
		 * non-repeating task.
		 */
		//private final long period;

		/** Sequence number to break ties FIFO */
		//private AtomicLong repeatCount = null;

		/** Sequence number to break ties FIFO */
		private final long sequenceNumber;

		/** The time the task is enabled to execute in nanoTime units */
		private long time;

		/** The actual task to be re-enqueued by reExecutePeriodic */
		RunnableScheduledFuture<V> outerTask = this;

		/** */
		public Trigger trigger;

		/**
		 * Creates a one-shot action with given nanoTime-based trigger.
		 */
		public GoTask1(Callable<V> callable, long ns) {
			super(callable);
			this.time = ns;
			this.sequenceNumber = sequencer.getAndIncrement();
		}

		/**
		 * Creates a one-shot action with given nanoTime-based trigger.
		 */
		public GoTask1(Runnable r, V result, CronExpression cronExpression) {
			super(r, result);
			Date date = cronExpression.getNextValidTimeAfter(new Date());
			long milliseconds = date.getTime();
			this.time = milliseconds;
			this.sequenceNumber = sequencer.getAndIncrement();
		}

		/**
		* Creates a one-shot action with given nanoTime-based trigger time.
		*/
		public GoTask1(Runnable r, V result, long ns) {
			super(r, result);
			this.time = ns;
			this.sequenceNumber = sequencer.getAndIncrement();
		}

		/**
		 * Creates a periodic action with given nano time and period.
		 */
		public GoTask1(Runnable r, V result, long ns, long period) {
			super(r, result);
			this.time = ns;
			//this.period = period;
			this.sequenceNumber = sequencer.getAndIncrement();
		}

		/**
		 * Creates a one-shot action with given nanoTime-based trigger.
		 */
		public GoTask1(Runnable r, V result, Trigger trigger) {
			super(r, result);
			this.sequenceNumber = sequencer.getAndIncrement();
			this.trigger = trigger;
			Date date = trigger.getNextFireTime();
			long milliseconds = date.getTime();
			this.time = milliseconds;
		}

		/**
		 * Runs a periodic task.
		 */
		@SuppressWarnings("synthetic-access")
		private void runPeriodic() {
			boolean ok = super.runAndReset();
			boolean down = isShutdown();
			// Reschedule if not cancelled and not shutdown or policy allows
			if (ok && (!down || (getContinueExistingPeriodicTasksAfterShutdownPolicy() && !isTerminating()))) {
				//				long p = period;
				//				if (p > 0) {
				//					time += p;
				//				} else {
				//					time = now() - p;
				//				}
				//	System.out.println("cc");
				//-------------------------------------
				Calendar cal = null;
				if (trigger.getCalendarName() != null) {
					//cal = retrieveCalendar(trigger.getCalendarName());
					cal = CalendarContent.getCalendar(trigger.getCalendarName());
					//if (cal == null)
					//	continue;
				}
				trigger.triggered(cal);
				//-------------------------------------
				Date date = trigger.getNextFireTime();
				//Date date = trigger.getFireTimeAfter(Calendar.getInstance().getTime());
				//Date date = cronExpression.getNextValidTimeAfter(Calendar.getInstance().getTime());
				if (date != null) {
					long milliseconds = date.getTime();
					this.time = milliseconds;
				} else {
					this.time = -1L;
				}
				//System.out.println("milliseconds: " + milliseconds);

				MyScheduledThreadPoolExecutor.super.getQueue().add(this);
			}
			// This might have been the final executed delayed
			// task.  Wake up threads to check.
			else if (down) {

			}
			//interruptIdleWorkers();
		}

		@Override
		public int compareTo(Delayed other) {
			//System.out.println("a");
			if (other == this) // compare zero ONLY if same object
				return 0;
			if (other instanceof GoTask) {
				GoTask<?> x = (GoTask<?>) other;
				long diff = time - x.time;
				if (diff < 0)
					return -1;
				else if (diff > 0)
					return 1;
				else if (sequenceNumber < x.sequenceNumber)
					return -1;
				else
					return 1;
			}
			long d = (getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS));
			return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
		}

		@Override
		public long getDelay(TimeUnit unit) {
			//System.out.println("time:" + time);
			//System.out.println("System.currentTimeMillis(): " + System.currentTimeMillis());
			//long delay = unit.convert(time - now(), TimeUnit.NANOSECONDS);
			long delay = time - System.currentTimeMillis();
			//System.out.println("delay:" + delay);
			return delay;
		}

		@Override
		public boolean isPeriodic() {
			return time > 0L;
			//return true;
		}

		@SuppressWarnings("synthetic-access")
		@Override
		public void run() {
			//System.out.println("b");
			if (isPeriodic())
				runPeriodic();
			else
				GoTask1.super.run();
		}
	}

	/** Base of nanosecond timings, to avoid wrapping */
	private static final long NANO_ORIGIN = System.nanoTime();
	/**
	* Sequence number to break scheduling ties, and in turn to
	* guarantee FIFO order among tied entries.
	*/
	static final AtomicLong sequencer = new AtomicLong(0);
	/**
	* False if should cancel/suppress periodic tasks on shutdown.
	*/
	private volatile boolean continueExistingPeriodicTasksAfterShutdown;

	/**
	* False if should cancel non-periodic tasks on shutdown.
	*/
	private volatile boolean executeExistingDelayedTasksAfterShutdown = true;

	/**
	 * Handler called when saturated or shutdown in execute.
	 */
	private volatile RejectedExecutionHandler handler;

	private boolean isPaused;

	private ReentrantLock pauseLock = new ReentrantLock();

	private Condition unpaused = pauseLock.newCondition();

	public MyScheduledThreadPoolExecutor(int corePoolSize) {
		super(corePoolSize, Integer.MAX_VALUE, 0L, TimeUnit.NANOSECONDS, new MyDelayedWorkQueue());
	}

	/**
	 * Creates a new ScheduledThreadPoolExecutor with the given
	 * initial parameters.
	 *
	 * @param corePoolSize the number of threads to keep in the pool,
	 * even if they are idle
	 * @param handler the handler to use when execution is blocked
	 * because the thread bounds and queue capacities are reached
	 * @throws IllegalArgumentException if <tt>corePoolSize &lt; 0</tt>
	 * @throws NullPointerException if handler is null
	 */
	public MyScheduledThreadPoolExecutor(int corePoolSize, RejectedExecutionHandler handler) {
		super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS, new MyDelayedWorkQueue(), handler);
	}

	/**
	 * Creates a new ScheduledThreadPoolExecutor with the given
	 * initial parameters.
	 *
	 * @param corePoolSize the number of threads to keep in the pool,
	 * even if they are idle
	 * @param threadFactory the factory to use when the executor
	 * creates a new thread
	 * @throws IllegalArgumentException if <tt>corePoolSize &lt; 0</tt>
	 * @throws NullPointerException if threadFactory is null
	 */
	public MyScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory) {
		super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS, new MyDelayedWorkQueue(), threadFactory);
	}

	/**
	* Creates a new ScheduledThreadPoolExecutor with the given
	* initial parameters.
	*
	* @param corePoolSize the number of threads to keep in the pool,
	* even if they are idle
	* @param threadFactory the factory to use when the executor
	* creates a new thread
	* @param handler the handler to use when execution is blocked
	* because the thread bounds and queue capacities are reached.
	* @throws IllegalArgumentException if <tt>corePoolSize &lt; 0</tt>
	* @throws NullPointerException if threadFactory or handler is null
	*/
	public MyScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
		super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS, new MyDelayedWorkQueue(), threadFactory, handler);
	}

	/**
	* Main execution method for delayed or periodic tasks.  If pool
	* is shut down, rejects the task. Otherwise adds task to queue
	* and starts a thread, if necessary, to run it.  (We cannot
	* prestart the thread to run the task because the task (probably)
	* shouldn't be run yet,) If the pool is shut down while the task
	* is being added, cancel and remove it if required by state and
	* run-after-shutdown parameters.
	*
	* @param task the task
	*/
	private void delayedExecute(RunnableScheduledFuture<?> task) {
		//		if (isShutdown())
		//			reject(task);
		//		else {
		//			super.getQueue().add(task);
		//			if (isShutdown() && !canRunInCurrentRunState(task.isPeriodic()) && remove(task))
		//				task.cancel(false);
		//			else
		//				prestartCoreThread();
		//		}
		if (isShutdown()) {
			reject(task);
			return;
		}
		// Prestart a thread if necessary. We cannot prestart it
		// running the task because the task (probably) shouldn't be
		// run yet, so thread will just idle until delay elapses.
		if (getPoolSize() < getCorePoolSize())
			prestartCoreThread();
		//System.out.println("c");
		super.getQueue().add(task);
		//System.out.println("d");
	}

	/**
	* Constrains the values of all delays in the queue to be within
	* Long.MAX_VALUE of each other, to avoid overflow in compareTo.
	* This may occur if a task is eligible to be dequeued, but has
	* not yet been, while some other task is added with a delay of
	* Long.MAX_VALUE.
	*/
	private long overflowFree(long delay) {
		Delayed head = (Delayed) super.getQueue().peek();
		if (head != null) {
			long headDelay = head.getDelay(TimeUnit.NANOSECONDS);
			if (headDelay < 0 && (delay - headDelay < 0))
				delay = Long.MAX_VALUE + headDelay;
		}
		return delay;
	}

	/**
	* Returns the trigger time of a delayed action.
	*/
	private long triggerTime(long delay, TimeUnit unit) {
		return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
	}

	@Override
	protected void beforeExecute(Thread t, Runnable r) {
		super.beforeExecute(t, r);
		pauseLock.lock();
		try {
			while (isPaused)
				unpaused.await();
		} catch (InterruptedException ie) {
			t.interrupt();
		} finally {
			pauseLock.unlock();
		}
	}

	/**
	* Method invoked when the Executor has terminated.  Default
	* implementation does nothing. Note: To properly nest multiple
	* overridings, subclasses should generally invoke
	* <tt>super.terminated</tt> within this method.
	*/
	@Override
	protected void terminated() {

	}

	/**
	* Modifies or replaces the task used to execute a runnable.
	* This method can be used to override the concrete
	* class used for managing internal tasks.
	* The default implementation simply returns the given task.
	*
	* @param runnable the submitted Runnable
	* @param task the task created to execute the runnable
	* @return a task that can execute the runnable
	* @since 1.6
	*/
	protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
		return task;
	}

	/**
	* Returns true if can run a task given current run state
	* and run-after-shutdown parameters.
	*
	* @param periodic true if this task periodic, false if delayed
	*/
	boolean canRunInCurrentRunState(boolean periodic) {
		//	return super.isRunningOrShutdown(periodic ? continueExistingPeriodicTasksAfterShutdown : executeExistingDelayedTasksAfterShutdown);
		return super.isShutdown();
	}

	/**
	 * Returns current nanosecond time.
	 */
	final long now() {
		return System.nanoTime() - NANO_ORIGIN;
	}

	/**
	* Returns the trigger time of a delayed action.
	*/
	long triggerTime(long delay) {
		return now() + ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
	}

	/**
	 * Gets the policy on whether to continue executing existing
	 * periodic tasks even when this executor has been
	 * <tt>shutdown</tt>. In this case, these tasks will only
	 * terminate upon <tt>shutdownNow</tt> or after setting the policy
	 * to <tt>false</tt> when already shutdown. This value is by
	 * default false.
	 *
	 * @return true if will continue after shutdown
	 * @see #setContinueExistingPeriodicTasksAfterShutdownPolicy
	 */
	public boolean getContinueExistingPeriodicTasksAfterShutdownPolicy() {
		return continueExistingPeriodicTasksAfterShutdown;
	}

	public boolean isPaused() {
		return isPaused;
	}

	public void pause() {
		pauseLock.lock();
		try {
			isPaused = true;
		} finally {
			pauseLock.unlock();
		}
	}

	/**
	* Wakes up all threads that might be waiting for tasks so they
	* can check for termination. Note: this method is also called by
	* ScheduledThreadPoolExecutor.
	*/
	//	void interruptIdleWorkers() {
	//		final ReentrantLock mainLock = this.mainLock;
	//		mainLock.lock();
	//		try {
	//			for (Worker w : workers)
	//				w.interruptIfIdle();
	//		} finally {
	//			mainLock.unlock();
	//		}
	//	}

	/**
	* Invokes the rejected execution handler for the given command.
	* Package-protected for use by ScheduledThreadPoolExecutor.
	*/
	public final void reject(Runnable command) {
		handler.rejectedExecution(command, this);
	}

	public void resume() {
		pauseLock.lock();
		try {
			isPaused = false;
			unpaused.signalAll();
		} finally {
			pauseLock.unlock();
		}
	}

	@Override
	public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
		return null;
	}

	@Override
	public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
		return null;
	}

	@Override
	public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
		return null;
	}

	/*
	 * 
	 */
	public ScheduledFuture<?> scheduleWithCronExp(Runnable command, CronExpression cronExpression) {
		if (command == null) {
			throw new NullPointerException();
		}
		if (cronExpression == null) {
			throw new IllegalArgumentException();
		}
		GoTask1<Void> sft = new GoTask1<Void>(command, null, cronExpression);
		RunnableScheduledFuture<Void> t = decorateTask(command, sft);
		sft.outerTask = t;
		delayedExecute(t);
		return t;
	}

	public ScheduledFuture<?> scheduleWithCronExp(Runnable command, Trigger trigger) {
		if (command == null) {
			throw new NullPointerException();
		}
		if (trigger == null) {
			throw new IllegalArgumentException();
		}
		GoTask1<Void> sft = new GoTask1<Void>(command, null, trigger);
		RunnableScheduledFuture<Void> t = decorateTask(command, sft);
		sft.outerTask = t;
		delayedExecute(t);
		return t;
	}

	@Override
	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
		if (command == null || unit == null) {
			throw new NullPointerException();
		}
		if (delay <= 0) {
			throw new IllegalArgumentException();
		}
		GoTask<Void> sft = new GoTask<Void>(command, null, triggerTime(initialDelay, unit), unit.toNanos(-delay));
		RunnableScheduledFuture<Void> t = decorateTask(command, sft);
		sft.outerTask = t;
		delayedExecute(t);
		return t;
	}
}
